package com.jingdianjichi.subject.infra.basic.ES.自定义ES_鸡翅;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.jingdianjichi.subject.infra.basic.ES.自定义ES_鸡翅.config.ESClusterConfig;
import com.jingdianjichi.subject.infra.basic.ES.自定义ES_鸡翅.config.ESConfigProperties;
import com.jingdianjichi.subject.infra.basic.ES.自定义ES_鸡翅.common.ESIndexInfo;
import com.jingdianjichi.subject.infra.basic.ES.自定义ES_鸡翅.common.ESSearchRequest;
import com.jingdianjichi.subject.infra.basic.ES.自定义ES_鸡翅.common.ESSourceData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.http.HttpHost;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.*;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.ScoreSortBuilder;
import org.springframework.stereotype.Component;

import org.elasticsearch.search.sort.SortOrder;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;

/**
 * 自定义客户端,我们封装一手亲
 */
@Component
@Slf4j
public class ESRestClient
{
    /**
     *客户端的初始化亲,封装链接
     */

    //集群客户端信心亲,后面想和谁交互就通过谁的集群名来获得对应的客户端,做各种各样的操作
    public static Map<String, RestHighLevelClient> clientMap=new HashMap<>();

    private static final RequestOptions COMMON_OPTIONS; //通用的参数我们初始化一下

    //初始化亲
    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();

        COMMON_OPTIONS=builder.build();
    }

    //集群信息亲
    @Resource
    private ESConfigProperties esConfigProperties;

    @PostConstruct              //这个注解就可以帮助我们进行初始化亲,这个也可以保证启动的时候加载,你也可以使用Bean的声明周期来的
    public void initialize(){
        List<ESClusterConfig> esClusterConfigs = esConfigProperties.getEsClusterConfigs();

        for (ESClusterConfig esConfig : esClusterConfigs) {
            log.info("initialize.config.name:{},node:{}",esConfig.getName(),esConfig.getNodes());

            //不同的集群返回不同的客户端
            RestHighLevelClient restHighLevelClient = initRestClient(esConfig);

            if (restHighLevelClient != null)
            {
                clientMap.put(esConfig.getName(),restHighLevelClient);
            }
            else
            {
                log.error("config.name:{},node:{}.initError",esConfig.getName(),esConfig.getNodes());
            }
        }
    }

    //初始化客户端,
    private RestHighLevelClient initRestClient(ESClusterConfig esClusterConfig){
        String[] ipPortArr = esClusterConfig.getNodes().split(",");     //我们一个索引库做多节点的,约定用,分隔

        //所有和ES的本质就是HTTP链接来的亲
        List<HttpHost> httpHostList=new ArrayList<>(ipPortArr.length);

        //把我们的数据处理一下亲,地址信心
        for (String ipPort : ipPortArr) {
            String[] ipPortInfo = ipPort.split(":");        //我们把地址和端口分开亲,0就是地址,1就是端口号来的

            //正确才处理下面的逻辑亲
            if (ipPortInfo.length==2){
                HttpHost httpHost = new HttpHost(ipPortInfo[0], NumberUtils.toInt(ipPortInfo[1]));

                httpHostList.add(httpHost);
            }
        }

        //和ES做一把交互
        HttpHost[] httpHosts=new HttpHost[httpHostList.size()];

        httpHostList.toArray(httpHosts);

        RestClientBuilder builder = RestClient.builder(httpHosts);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);

        return restHighLevelClient;
    }


    /**
     * 封装常规操作
     * QueryBuilders是简单操作来的
     * BoolQueryBuilder是复杂查询来的
     */

    //获得客户端
    private static RestHighLevelClient getClient(String clusterName){
        return clientMap.get(clusterName);
    }


    //新增文档,要知道集群的哪个索引库,和你要操作的元数据
    public static boolean insertDoc(ESIndexInfo esIndexInfo, ESSourceData sourceData){
        try {
            IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName());       //传递索引库的名称
            indexRequest.source(sourceData.getData());                                      //设置数据亲

            indexRequest.id(sourceData.getDocId());                                         //设置id

            //获得对应集群的客户端
            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            client.index(indexRequest,COMMON_OPTIONS);
            //client.index(indexRequest,RequestOptions.DEFAULT);  之前我们是这样写的

            return true;
        }catch (Exception e){
            log.error("insertDoc,exception:{}",e.getMessage(),e);
        }

        return false;
    }

    //删除文档全部
    public static boolean deletedALLDoc(ESIndexInfo esIndexInfo){
        try {
            DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(esIndexInfo.getIndexName());

            //删除索引库中所有的消息
            deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery());

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            BulkByScrollResponse response = client.deleteByQuery(deleteByQueryRequest, COMMON_OPTIONS);

            long deleted = response.getDeleted();

            log.info("deletedALLDoc.size:{}",deleted);

            return true;
        }catch (Exception e){
            log.error("deletedALLDoc.exception:{}",e.getMessage(),e);
        }

        return false;
    }

    //删除文档指定的亲
    public static boolean deletedDocByDocId(ESIndexInfo esIndexInfo,String docId){
        try {
            //删除单个的
            DeleteRequest deleteRequest = new DeleteRequest(esIndexInfo.getIndexName());
            deleteRequest.id(docId);

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            DeleteResponse response = client.delete(deleteRequest, COMMON_OPTIONS);

            log.info("deletedALLDoc.size:{}", JSON.toJSONString(response));

            return true;
        }catch (Exception e){
            log.error("deletedDocByDocId.exception:{}",e.getMessage(),e);
        }

        return false;
    }


    //查看文档是否存在
    public static boolean isExistDocById(ESIndexInfo esIndexInfo,String docId){
        try {
            //看看文档是否存在
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            boolean exists = client.exists(getRequest, COMMON_OPTIONS);

            return exists;
        }catch (Exception e){
            log.error("isExistDocById.exception:{}",e.getMessage(),e);
        }

        return false;
    }


    //获得指定文档的数据
    public static Map<String,Object> getDocById(ESIndexInfo esIndexInfo,String docId){
        try {
            //获得指定文档的数据
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            GetResponse response = client.get(getRequest, COMMON_OPTIONS);

            Map<String, Object> source = response.getSource();      //人家返回的本身就是这个所以 我们存储的时候也是这样存储的,你也可以返回字符串的

            return source;
        }catch (Exception e){
            log.error("getDocById.exception:{}",e.getMessage(),e);
        }

        //没有直接返回空的
        return null;
    }


    //根据指定的字段去索引,参数就是你指定的字段去查的
    public static Map<String,Object> getDocById(ESIndexInfo esIndexInfo,String docId,String[] fields){
        try {
            //获得指定文档的数据
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);

            //这里就是true包含我们指定的字段,null就是我们不排序啥字段来的
            FetchSourceContext fetchSourceContext = new FetchSourceContext(true, fields, null);
            getRequest.fetchSourceContext(fetchSourceContext);          //我们之前在学的时候是手动拼接字符串的,这样非常的专业亲

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            GetResponse response = client.get(getRequest,COMMON_OPTIONS);

            Map<String, Object> source = response.getSource();

            return source;
        }catch (Exception e){
            log.error("getDocById.exception:{}",e.getMessage(),e);
        }

        //没有直接返回空的
        return null;
    }

    /**
     * 下面就是搜索的方法了
     */
    //参数就是我们的封装的请求参数来的
    public static SearchResponse searchWithTermQuery(ESIndexInfo esIndexInfo, ESSearchRequest searchRequest){
        try {
            BoolQueryBuilder bq = searchRequest.getBq();        //获得指定的查询条件,用于复杂查询亲,里面包含简单查询
            String[] fields = searchRequest.getFields();        //获得指定的查询字段
            int from = searchRequest.getFrom();                 //页数
            int size = searchRequest.getSize();                 //条数
            Long minutes = searchRequest.getMinutes();          //快照留存时间
            Boolean needScroll = searchRequest.getNeedScroll(); //是否要快照
            String sortName = searchRequest.getSortName();      //排序的字段
            SortOrder sortOrder = searchRequest.getSortOrder(); //排序的规则

            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();        //用于搜索的

            //设置查询条件
            searchSourceBuilder.query(bq);

            //设置查询的字段
            searchSourceBuilder.fetchSource(fields,null).from(from).size(size);     //设置查询的字段

            //设置高亮
            if (Objects.nonNull(searchRequest.getHighlightBuilder())){
                searchSourceBuilder.highlighter(searchRequest.getHighlightBuilder());
            }

            //设置排序字段
            if (StringUtils.isNotBlank(sortName)){
                searchSourceBuilder.sort(sortName);
            }

            //设置排序规则
            searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));

            /*
                if (sortOrder!=null){
                    searchSourceBuilder.sort(String.valueOf(sortOrder));
                }
             */

            SearchRequest request = new SearchRequest();

            request.searchType(SearchType.DEFAULT);
            request.indices(esIndexInfo.getIndexName());
            request.source(searchSourceBuilder);
            //快照
            if (needScroll){
                Scroll scroll = new Scroll(TimeValue.timeValueMinutes(minutes));
                request.scroll(scroll);
            }

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());
            SearchResponse response = client.search(request, COMMON_OPTIONS);

            return response;
        }catch (Exception e){
            log.error("getDocById.exception:{}",e.getMessage(),e);
        }

        //没有直接返回空的
        return null;
    }


    //更新文档
    public static boolean updateDoc(ESIndexInfo esIndexInfo,ESSourceData sourceData){
        try {
            UpdateRequest updateRequest = new UpdateRequest();

            updateRequest.index(esIndexInfo.getIndexName());    //设置索引库
            updateRequest.id(sourceData.getDocId());
            updateRequest.doc(sourceData.getData());

            RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

            client.update(updateRequest,COMMON_OPTIONS);

            return true;

        }catch (Exception e){
            log.error("updateDoc.exception:{}",e.getMessage(),e);
        }

        return false;
    }

    //批量更新亲
    public static boolean batchUpdateDoc(ESIndexInfo esIndexInfo,List<ESSourceData> sourceDataList){
        try {
            boolean flag=false;

            BulkRequest bulkRequest = new BulkRequest();        //BulkRequest是用来执行批量操作的请求类

            for (ESSourceData esSourceData : sourceDataList) {
                String docId = esSourceData.getDocId();
                if (StringUtils.isNotBlank(docId)){
                    UpdateRequest updateRequest = new UpdateRequest();
                    updateRequest.index(esIndexInfo.getIndexName());
                    updateRequest.id(esSourceData.getDocId());
                    updateRequest.doc(esSourceData.getData());

                    bulkRequest.add(updateRequest);
                    flag=true;
                }
            }

            if (flag){
                RestHighLevelClient client = getClient(esIndexInfo.getClusterName());

                BulkResponse bulk = client.bulk(bulkRequest, COMMON_OPTIONS);
                if (bulk.hasFailures()){
                    return false;       //这里就是看看是不是所有的都更新成功了亲
                }
            }

            return true;

        }catch (Exception e){
            log.error("batchUpdateDoc.exception:{}",e.getMessage(),e);
        }

        return false;
    }

    //根据查询的条件来更新
    public static boolean updateByQuery(ESIndexInfo esIndexInfo, QueryBuilder queryBuilder, Script script, int batchSize) {
        if (log.isInfoEnabled()) {
            log.info("updateByQuery.indexName:" + esIndexInfo.getIndexName());
        }
        try {
            UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(esIndexInfo.getIndexName());
            updateByQueryRequest.setQuery(queryBuilder);
            updateByQueryRequest.setScript(script);
            updateByQueryRequest.setBatchSize(batchSize);
            updateByQueryRequest.setAbortOnVersionConflict(false);
            BulkByScrollResponse response = getClient(esIndexInfo.getClusterName()).updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
            List<BulkItemResponse.Failure> failures = response.getBulkFailures();
        } catch (Exception e) {
            log.error("updateByQuery.error", e);
        }
        return true;
    }

    /**
     * 分词方法
     */
    public static List<String> getAnalyze(ESIndexInfo esIndexInfo, String text) throws Exception {
        List<String> list = new ArrayList<String>();
        Request request = new Request("GET", "_analyze");
        JSONObject entity = new JSONObject();
        entity.put("analyzer", "ik_smart");
        entity.put("text", text);
        request.setJsonEntity(entity.toJSONString());
        Response response = getClient(esIndexInfo.getClusterName()).getLowLevelClient().performRequest(request);
        JSONObject tokens = JSONObject.parseObject(EntityUtils.toString(response.getEntity()));
        JSONArray arrays = tokens.getJSONArray("tokens");
        for (int i = 0; i < arrays.size(); i++) {
            JSONObject obj = JSON.parseObject(arrays.getString(i));
            list.add(obj.getString("token"));
        }
        return list;
    }

}
